﻿using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.Text;

namespace Common
{
    public partial class DbUtil
    {
        /// <summary>
        /// 
        /// </summary>
        /// <param name="connStr"></param>
        /// <param name="schemaName"></param>
        /// <param name="tblName"></param>
        /// <param name="partitionField"></param>
        /// <param name="outputTableCount"></param>
        /// <param name="outputTablePrefix"></param>
        public static void PartitionTableOnFieldUsingStoredProc(
            string connStr, string schemaName, string tblName,
            string partitionField, int outputTableCount, string outputTablePrefix)
        {
            try
            {
                string spName = "TMP_PARTITION_TABLE_SP";
                string spBody = @"
CREATE PROCEDURE TMP_PARTITION_TABLE_SP
(@SCHEMA_NAME NVARCHAR(50),
@TABLE_NAME NVARCHAR(100),
@PREFIX NVARCHAR(100),
@PK_FIELD NVARCHAR(100),
@PARTITION_FIELD NVARCHAR(100),
@OUTPUT_COUNT INT) AS 
BEGIN
--DECLARE @SCHEMA_NAME NVARCHAR(50);
--SET @SCHEMA_NAME='ORG';
--DECLARE @TABLE_NAME NVARCHAR(100);
--SET @TABLE_NAME='ORGS';
--DECLARE @PREFIX NVARCHAR(100);
--SET @PREFIX='ORG_PART_';
--DECLARE @PK_FIELD NVARCHAR(50);
--SET @PK_FIELD='ORG_ID';
--DECLARE @PARTITION_FIELD NVARCHAR(100);
--SET @PARTITION_FIELD='ORG_NAME';
--DECLARE @OUTPUT_COUNT INT;
--SET @OUTPUT_COUNT=40;
DECLARE @SQL NVARCHAR(1000);
DECLARE @INPUT NVARCHAR(1000);
SET @INPUT='@RECORD_COUNT INT OUTPUT';
SET @SQL='SELECT @RECORD_COUNT=COUNT(*) FROM '+@SCHEMA_NAME+'.'+@TABLE_NAME+';';
DECLARE @REC_COUNT INT;
PRINT @SQL;
EXEC SP_EXECUTESQL @SQL, @INPUT, @REC_COUNT OUTPUT;
DECLARE @PARTITION_SIZE INT;
SET @PARTITION_SIZE=@REC_COUNT/@OUTPUT_COUNT;

SET @SQL=
'IF EXISTS (SELECT * FROM SYS.ALL_OBJECTS WHERE NAME=''TMP_PROCESSED'' AND SCHEMA_ID IN (SELECT SCHEMA_ID FROM sys.schemas WHERE name='''+@SCHEMA_NAME+'''))
	DROP TABLE '+@SCHEMA_NAME+'.TMP_PROCESSED;';
PRINT @SQL;
EXEC SP_EXECUTESQL @SQL;

SET @SQL=
'CREATE TABLE '+@SCHEMA_NAME+'.TMP_PROCESSED ('+@PK_FIELD+' INT);';
PRINT @SQL;
EXEC SP_EXECUTESQL @SQL;

SET @SQL=
'CREATE INDEX IX_TMP_PROCESSED_ID ON '+@SCHEMA_NAME+'.TMP_PROCESSED ('+@PK_FIELD+');';
PRINT @SQL;
EXEC SP_EXECUTESQL @SQL;

DECLARE @PART_NAME NVARCHAR(50);
DECLARE @PART INT;
SET @PART=0;
WHILE (@PART<=@OUTPUT_COUNT)
BEGIN
	SET @PART=@PART+1;
	SET @PART_NAME=@PREFIX+CAST(@PART AS VARCHAR);

	SET @SQL=
	'IF EXISTS (SELECT * FROM SYS.ALL_OBJECTS WHERE NAME='''+@PART_NAME+''' AND SCHEMA_ID IN (SELECT SCHEMA_ID FROM SYS.SCHEMAS WHERE NAME='''+@SCHEMA_NAME+''')) 
	DROP TABLE '+@SCHEMA_NAME+'.'+@PART_NAME+';';
	PRINT @SQL;
	EXEC SP_EXECUTESQL @SQL;

	SET @SQL='SELECT TOP ' + CAST(@PARTITION_SIZE AS VARCHAR) + 
	' * INTO ' + @SCHEMA_NAME+'.'+@PART_NAME + ' FROM '+@SCHEMA_NAME+'.'+@TABLE_NAME+' WHERE '+@PK_FIELD+' NOT IN ('+
	'SELECT '+@PK_FIELD+' FROM '+@SCHEMA_NAME+'.TMP_PROCESSED) ORDER BY '+@PARTITION_FIELD+';';
	PRINT @SQL;
	EXEC SP_EXECUTESQL @SQL;

	SET @SQL='INSERT INTO '+@SCHEMA_NAME+'.TMP_PROCESSED('+@PK_FIELD+') SELECT '+@PK_FIELD+' FROM '+@SCHEMA_NAME+'.'+@PART_NAME+';';
	PRINT @SQL;
	EXEC SP_EXECUTESQL @SQL;
END

SET @PART=0;
WHILE (@PART<=@OUTPUT_COUNT)
BEGIN
	SET @PART=@PART+1;
	IF(@PART <=@OUTPUT_COUNT)
	BEGIN
		SET @PART_NAME=@PREFIX+CAST(@PART AS VARCHAR);
		DECLARE @NEXT_PART_NAME NVARCHAR(100);
		SET @NEXT_PART_NAME=@PREFIX+CAST((@PART+1) AS VARCHAR);
		
		SET @SQL=	
		'IF EXISTS (SELECT * FROM sys.identity_columns WHERE OBJECT_ID IN (
		SELECT OBJECT_ID FROM sys.all_objects WHERE name='''+@NEXT_PART_NAME+''' AND SCHEMA_ID IN (
		SELECT SCHEMA_ID FROM sys.schemas WHERE name='''+@SCHEMA_NAME+''')))
		SET IDENTITY_INSERT '+@SCHEMA_NAME+'.'+@NEXT_PART_NAME+' ON; 
		INSERT INTO '+@SCHEMA_NAME+'.'+@NEXT_PART_NAME + ' SELECT * FROM '+@SCHEMA_NAME+'.'+@PART_NAME+ 
		' WHERE '+@PARTITION_FIELD+' IN (SELECT TOP 1 '+@PARTITION_FIELD+' FROM '+@SCHEMA_NAME+'.'+@NEXT_PART_NAME+
		' ORDER BY '+@PARTITION_FIELD+' ASC);';
		PRINT @SQL;
		EXEC SP_EXECUTESQL @SQL;
		SET @SQL='DELETE FROM '+@SCHEMA_NAME+'.'+@PART_NAME + ' WHERE '+@PK_FIELD+
		' IN (SELECT TOP 1 '+@PK_FIELD+' FROM '+@SCHEMA_NAME+'.'+@NEXT_PART_NAME+' ORDER BY '+@PK_FIELD+' ASC);';
		PRINT @SQL;
		EXEC SP_EXECUTESQL @SQL;
	END
END
END";
                bool created = DbUtil.NewStoredProc(connStr, spName, spBody);
                if (!created)
                    throw new Exception("Unable to create partition stored procedure");
                Dictionary<string, object> spParameters = new Dictionary<string, object>();
                List<string> pkFields = DbUtil.GetTablePrimaryKeys(connStr, schemaName, tblName);
                if (pkFields == null || pkFields.Count != 1)
                    throw new Exception("Unable to find single pk field from table");
                spParameters.Add("SCHEMA_NAME", schemaName);
                spParameters.Add("TABLE_NAME", tblName);
                spParameters.Add("PREFIX", outputTablePrefix);
                spParameters.Add("PK_FIELD", pkFields[0]);
                spParameters.Add("PARTITION_FIELD", partitionField);
                spParameters.Add("OUTPUT_COUNT", outputTableCount);
                int timeoutSeconds = 60 * 60; // 1 hr
                DbUtil.ExecuteStoredProcedure(connStr, spName, spParameters, timeoutSeconds);
            }
            catch (Exception ex)
            {
                ExceptionHandler.HandlesException(ex);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="connStr"></param>
        /// <param name="schemaName"></param>
        /// <param name="tblName"></param>
        /// <param name="partitionField"></param>
        /// <param name="outputTableCount"></param>
        /// <param name="outputTablePrefix"></param>
        public static void PartitionTableOnField(
            string connStr, string schemaName, string tblName,
            string partitionField, int outputTableCount, string outputTablePrefix)
        {
            int recCount = DbUtil.GetTableRecordCount(DataProviderType.MSSQL,
                                                      connStr, schemaName, tblName);
            int binSize = recCount/outputTableCount;
            TableColumn partitionColumn = DbUtil.GetTableColumn(
                DataProviderType.MSSQL, connStr, schemaName, tblName, partitionField);

            SqlConnection conn=new SqlConnection(connStr);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                string sql = "select [{2}], count (*) as Ct from {0}.[{1}] where [{2}] is not null group by [{2}] order by [{2}]";
                cmd.CommandText = string.Format(sql, schemaName, tblName, partitionField);
                cmd.CommandType = CommandType.Text;
                SqlDataReader reader = cmd.ExecuteReader();
                Dictionary<object, int> distributions=new Dictionary<object, int>();
                Dictionary<int, object> partFields1 = new Dictionary<int, object>();
                Dictionary<int, object> partFields2 = new Dictionary<int, object>();
                while (reader.Read())
                {
                    object val = reader[partitionField];
                    int ct = int.Parse(reader["ct"].ToString());
                    distributions.Add(val, ct);
                }
                reader.Close();

                int processed = 0;
                int partNum = 0;
                object prevObj = null;
                object nextObj = null;
                foreach(object distVal in distributions.Keys)
                {
                    if(prevObj==null)
                        prevObj = distVal;
                    processed += distributions[distVal];
                    if(processed >=binSize)
                    {
                        partNum++;
                        partFields2.Add(partNum, distVal);
                        partFields1.Add(partNum, prevObj);
                        processed = 0;
                        prevObj = null;
                    }
                    nextObj = distVal;
                }
                if(processed>0)
                {
                    partNum++;
                    partFields2.Add(partNum, nextObj);
                    partFields1.Add(partNum, prevObj);
                }
                for(int i=1; i<=partNum;i++)
                {
                    string partTblName = outputTablePrefix + i.ToString();
                    string dropSql =
                        @"IF EXISTS (SELECT * FROM [sys].[all_objects] 
WHERE 
	NAME='{1}' 
	AND [schema_id] IN (SELECT [schema_id] FROM [sys].[schemas] WHERE NAME='{0}') 
	AND TYPE='U') 
DROP TABLE {0}.[{1}];";
                    cmd.CommandText = string.Format(dropSql, schemaName, partTblName);
                    cmd.ExecuteNonQuery();

                    object fromVal = partFields1[i];
                    object toVal = partFields2[i];
                    string whereSql = string.Format(
                        "[{0}] >= {1} and [{0}] <= {2}",
                        partitionField,
                        FieldDataType.ToMssqlDataValue(partitionColumn.DbType, fromVal.ToString()),
                        FieldDataType.ToMssqlDataValue(partitionColumn.DbType, toVal.ToString()));
                    string insertSql = string.Format(
                        "select * into {0}.[{1}] from {0}.[{2}] where {3}",
                        schemaName, partTblName, tblName, whereSql);
                    cmd.CommandText = insertSql;
                    cmd.ExecuteNonQuery();
                }
            }
            catch(Exception ex)
            {
                ExceptionHandler.HandlesException(ex);
            }
            finally
            {
                conn.Close();
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="connStr"></param>
        /// <param name="schemaName"></param>
        /// <param name="tblName"></param>
        /// <param name="partitionFields"></param>
        /// <param name="recordPerTable"></param>
        /// <param name="tablePrefix"></param>
        public static int PartitionTableOnFields(
            string connStr, string schemaName, string tblName,
            List<string> partitionFields, int recordPerTable, 
            string tablePrefix)
        {
            string partitionFieldSql = string.Empty;
            foreach(string partitionField in partitionFields)
            {
                if (partitionFieldSql.Length > 0)
                    partitionFieldSql += "+'-'+";
                partitionFieldSql += string.Format("cast(ISNULL([{0}],'') as varchar)", partitionField);
            }
            int partNum = 0;
            SqlConnection conn = new SqlConnection(connStr);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                string sql = string.Format(
                    "select {0} as PartitionField, count(*) as ct from [{1}].[{2}] group by {0} order by {0}",
                    partitionFieldSql, schemaName, tblName);
                cmd.CommandText = sql;
                cmd.CommandType = CommandType.Text;
                cmd.CommandTimeout = 60*30;// 30 min 
                SqlDataReader reader = cmd.ExecuteReader();
                Dictionary<string, int> distributions = new Dictionary<string, int>();
                Dictionary<int, string> partFields1 = new Dictionary<int, string>();
                Dictionary<int, string> partFields2 = new Dictionary<int, string>();
                while (reader.Read())
                {
                    string val = (reader["PartitionField"] == null || reader["PartitionField"] == DBNull.Value)
                                     ? null
                                     : reader["PartitionField"].ToString();
                    int ct = int.Parse(reader["ct"].ToString());
                    if(!string.IsNullOrEmpty(val))
                    {
                        val = val.Replace("'", "");
                        if (!distributions.ContainsKey(val))
                            distributions.Add(val, ct);
                    }
                }
                reader.Close();

                int processed = 0;
                
                string prevObj = null;
                string nextObj = null;
                foreach (string distVal in distributions.Keys)
                {
                    if (prevObj == null)
                        prevObj = distVal;
                    processed += distributions[distVal];
                    if (processed >= recordPerTable)
                    {
                        partNum++;
                        partFields2.Add(partNum, distVal);
                        partFields1.Add(partNum, prevObj);
                        processed = 0;
                        prevObj = null;
                    }
                    nextObj = distVal;
                }
                if (processed > 0)
                {
                    partNum++;
                    partFields2.Add(partNum, nextObj);
                    partFields1.Add(partNum, prevObj);
                }
                for (int i = 1; i <= partNum; i++)
                {
                    string partTblName = tablePrefix + i.ToString();
                    string dropSql =
                        @"IF EXISTS (SELECT * FROM [sys].[all_objects] 
WHERE 
	NAME='{1}' 
	AND [schema_id] IN (SELECT [schema_id] FROM [sys].[schemas] WHERE NAME='{0}') 
	AND TYPE='U') 
DROP TABLE {0}.[{1}];";
                    cmd.CommandText = string.Format(dropSql, schemaName, partTblName);
                    cmd.ExecuteNonQuery();

                    string fromVal = partFields1[i];
                    string toVal = partFields2[i];
                    string whereSql = string.Format(
                        "{0} >= '{1}' and {0} <= '{2}'",
                        partitionFieldSql,
                        fromVal,
                        toVal);
                    string insertSql = string.Format(
                        "select * into {0}.[{1}] from {0}.[{2}] where {3}",
                        schemaName, partTblName, tblName, whereSql);
                    cmd.CommandText = insertSql;
                    cmd.CommandTimeout = 60*30;
                    cmd.ExecuteNonQuery();
                }
            }
            catch (Exception ex)
            {
                ExceptionHandler.HandlesException(ex);
            }
            finally
            {
                conn.Close();
            }
            return partNum;
        }

        //private static string CreateTableAndInsertStoredProcedure(string connStr, string schemaName, string tgtTableName, TableColumn[] cols)
        //{
        //    DbUtil.NewTable(DataProviderType.MSSQL, connStr, schemaName, tgtTableName, cols);
        //    string spName = string.Format("bulkInsert_{0}", tgtTableName);
        //    string spBody = StoredProcUtil.CreateStoredProcedureStatementForInsert(DataProviderType.MSSQL, connStr,
        //                                                                           schemaName, tgtTableName, cols);
        //    DbUtil.NewStoredProc(connStr, spName, spBody);
        //    return spName;
        //}

        /// <summary>
        /// 
        /// </summary>
        /// <param name="connStr"></param>
        /// <param name="schemaName"></param>
        /// <param name="tblName"></param>
        /// <param name="dt"></param>
        /// <param name="cols"></param>
        public static void SaveTableData(
            string connStr, string schemaName, string tblName,
            DataTable dt, List<TableColumn> cols)
        {
            SqlConnection conn=new SqlConnection(connStr);
            try
            {
                conn.Open();
                string sp_Name = string.Format("tmp_BulkInsert_{0}", tblName);
                string sp_Body = StoredProcUtil.CreateStoredProcedureStatementForInsert(
                    DataProviderType.MSSQL, sp_Name, schemaName, tblName, cols.ToArray());
                DbUtil.NewStoredProc(connStr, sp_Name, sp_Body);

                Dictionary<string,SqlParameter> sqlParams=new Dictionary<string, SqlParameter>();
                foreach(TableColumn col in cols)
                {
                    SqlParameter sqlParam=new SqlParameter("@"+col.ColumnName, DBNull.Value);
                    sqlParams.Add(col.ColumnName, sqlParam);
                }
                SqlCommand cmd = conn.CreateCommand();
                cmd.CommandText = sp_Name;
                cmd.CommandType = CommandType.StoredProcedure;
                foreach(DataRow dr in dt.Rows)
                {
                    cmd.Parameters.Clear();
                    foreach(string colName in sqlParams.Keys)
                    {
                        SqlParameter sqlParam = sqlParams[colName];
                        if(dr[colName]!=null && dr[colName]!=DBNull.Value)
                            sqlParam.Value = dr[colName];
                        else
                            sqlParam.Value = DBNull.Value;
                        cmd.Parameters.Add(sqlParam);
                    }
                    cmd.ExecuteNonQuery();
                }
            }
            catch(Exception ex)
            {
                ExceptionHandler.HandlesException(ex);
            }
            finally
            {
                conn.Close();
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="connStr"></param>
        /// <param name="schemaName"></param>
        /// <param name="tblName"></param>
        /// <returns></returns>
        public static DataTable GetTableData(string connStr, string schemaName, string tblName)
        {
            DataTable dtOutput = null;
            SqlConnection conn = new SqlConnection(connStr);
            try
            {
                conn.Open();
                string sql = string.Format("select * from {0}.[{1}]", schemaName, tblName);
                SqlDataAdapter daSelect = new SqlDataAdapter(sql, conn);
                DataSet dsOutput = new DataSet();
                daSelect.Fill(dsOutput, tblName);
                dtOutput = dsOutput.Tables[0];
            }
            catch (Exception ex)
            {
                ExceptionHandler.HandlesException(ex);
            }
            finally
            {
                conn.Close();
            }
            return dtOutput;
        }

        #region constraints
        /// <summary>
        /// 
        /// </summary>
        /// <param name="connStr"></param>
        /// <param name="foreignKeyName"></param>
        /// <param name="foreignSchema"></param>
        /// <param name="foreignTable"></param>
        /// <param name="foreignField"></param>
        /// <param name="primarySchema"></param>
        /// <param name="primaryTable"></param>
        /// <param name="primaryField"></param>
        public static void CreateForeignKey(
            string connStr, string foreignKeyName,
            string foreignSchema, string foreignTable, string foreignField,
            string primarySchema, string primaryTable, string primaryField)
        {
            string sql =
                @"IF NOT EXISTS (SELECT * FROM sys.foreign_keys 
       WHERE object_id = OBJECT_ID(N'[{0}].[{6}]')
       AND parent_object_id = OBJECT_ID(N'[{0}].[{1}]'))
ALTER TABLE [{0}].[{1}]  WITH CHECK ADD  
       CONSTRAINT [FK_OnsiteCourse_Course] FOREIGN KEY([{2}])
REFERENCES [{3}].[{4}] ([{5}])
GO
ALTER TABLE [{0}].[{1}] CHECK 
       CONSTRAINT [{6}]
GO";
            sql = string.Format(
                sql, foreignSchema, foreignTable, foreignField,
                primarySchema, primaryTable, primaryField, foreignKeyName);
            DbUtil.ExecuteSqlCommand(connStr, sql);
        }
        #endregion
    }
}
